package com.huwaiwai.thread.thread;

import com.google.common.collect.Lists;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.huwaiwai.thread.task.TaskFunction;
import com.huwaiwai.thread.task.TaskRequest;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.concurrent.*;

/**
 * 线程池
 * Created by huweiwei on 2017/4/10.
 */
public class MyThreadPool {
    private static final int DEFAULT_MIN_THREAD_COUNT = 16;
    private static final int DEFAULT_MAX_THREAD_COUNT = 64;
    private static final int DEFAULT_KEEP_ALIVE_TIME = 60;
    private static final int DEFAULT_QUEUE_SIZE = 32;
    private static final long DEFAULT_THREAD_PROCESS_TIME_OUT = 60000L;//60s
    private static final long DEFAULT_FUTURE_GET_TIME_OUT = 10L; //10ms

    private int corePoolSize = DEFAULT_MIN_THREAD_COUNT;
    private int maximumPoolSize = DEFAULT_MAX_THREAD_COUNT;
    private int keepAliveTime = DEFAULT_KEEP_ALIVE_TIME;
    private int queueSize = DEFAULT_QUEUE_SIZE;

    private BlockingQueue<Runnable> workQueue;

    private RejectedExecutionHandler handler;
    private ThreadFactory threadFactory;
    private ThreadPoolExecutor threadPoolExecutor;
    private ListeningExecutorService service;

    @PostConstruct
    public void initialize() {
        System.out.println("MyThreadPool-线程池初始化");
        this.workQueue = new ArrayBlockingQueue(queueSize);
        this.handler = new ThreadPoolExecutor.CallerRunsPolicy();
        this.threadFactory = new NameThreadFactory("WW-Parallel-Processor", true);
        this.threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.SECONDS, workQueue, threadFactory, handler);
        this.service = MoreExecutors.listeningDecorator(threadPoolExecutor);
    }

    @PreDestroy
    public void stop() {
        service.shutdownNow();
        workQueue.clear();
    }
    public <V> List<V> execute(TaskRequest taskRequest) {
        return execute(taskRequest, DEFAULT_THREAD_PROCESS_TIME_OUT);
    }
    public <V> List<V> execute(TaskRequest taskRequest, long processTimeOut) {
        int taskCount = taskRequest.getTaskCount();
//        System.out.println("开始执行并发请求，数量：" + taskCount);
        CountDownLatch latch = new CountDownLatch(taskCount);
        List<ListenableFuture<V>> futureList = Lists.newArrayList();
        for (int i = 0; i < taskCount; i++) {
            int index = i;
            ListenableFuture<V> futureTaskResult = service.submit(() -> {
                long start = System.currentTimeMillis();
                try {
                    TaskFunction<V> taskFunction = taskRequest.getTaskList().get(index);
                    V result = taskFunction.apply();
//                    System.out.println(Thread.currentThread().getName() + "执行第" + index + "个任务。耗时：" + (System.currentTimeMillis() - start) + " ms");
                    return result;
                } catch (Throwable e) {
                    System.out.println(Thread.currentThread().getName() + "执行第" + index + "个任务。失败");
                    e.printStackTrace();
                    throw new RuntimeException();
                } finally {
                    latch.countDown();
                }
            });
            futureList.add(futureTaskResult);
        }
        ListenableFuture<List<V>> finalFuture;
        if (taskRequest.isIgnoreError()) {
            finalFuture = Futures.successfulAsList(futureList);
        } else {
            finalFuture = Futures.allAsList(futureList);
        }
        if (taskRequest.getFutureCallback() != null) {
            Futures.addCallback(finalFuture, taskRequest.getFutureCallback());
        }
        try {
            latch.await(processTimeOut, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        List<V> taskResultList = Lists.newArrayList();
        try {
            taskResultList = finalFuture.get(DEFAULT_FUTURE_GET_TIME_OUT, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
        return taskResultList;
    }
}
